/*
 *  Filename:lsv_log_thread.c
 *  Description:
 *
 *  Created on: 2017年3月4日
 *  Author: Asdf(825674301)
 */
#include "config.h"

#define DBG_SUBSYS S_LIBSTORAGE

#include "core.h"
#include "volume_proto.h"

#include "lsv_log.h"
#include "lsv_gc_internal.h"
#include "dbg.h"

extern uint32_t lsv_feature;

static int __do_log_gc(void *arg) {
        lsv_gc_taskctx_t *ctx = arg;
        lsv_volume_proto_t *lsv_info = ctx->lsv_info;
        lsv_gc_arg_t *gc_arg = ctx->gc_arg;

        int ret = 0;

        DINFO_NOP("enter ino %llu\n", (LLU )lsv_info->ino);

        if (LSV_GC_ARG_TPYE_GC == gc_arg->msg_type) {
//                ret = lsv_log_gc_(lsv_info, gc_arg->type);
                assert(0);
        } else if (LSV_GC_ARG_TPYE_VALUEUP == gc_arg->msg_type) {
                ret = lsv_gc_do_valueup(lsv_info, gc_arg->chunk_id, gc_arg->chunk_page_off);
        } else {
                DERROR("log_gc %llu gc_arg err, err: %d,msg_type:%u\n", (LLU )lsv_info->ino, ret, gc_arg->msg_type);
                assert(0);
        }

        if (ret < 0) {
                DERROR("log_gc %llu work err, err: %d\n", (LLU )lsv_info->ino, ret);
        }

        if (gc_arg->is_free) {
                free(gc_arg);
                ctx->gc_arg = NULL;
        }

        DINFO_NOP("exit ino %llu\n", (LLU )lsv_info->ino);

        return 0;
}

int lsv_gc_offer(lsv_volume_proto_t *lsv_info, lsv_gc_arg_t *gc_arg) {
        int ret;
        lsv_log_info_t *log_info = lsv_info->log_info;
        lsv_gc_context_t *gc_ctx = log_info->gc_context;
        lsv_gc_taskctx_t *ctx;

        if (lsv_feature & LSV_FEATURE_GC) {
                // @malloc ERR1
                ret = ymalloc((void **) &ctx, sizeof(lsv_gc_taskctx_t));
                if (unlikely(ret))
                        GOTO(ERR0, ret);

                ctx->lsv_info = lsv_info;
                ctx->gc_arg = gc_arg;

                // 顺序处理
                ret = co_mq_offer(&gc_ctx->gc_queue, "log_gc", __do_log_gc, ctx, 1);
                if (unlikely(ret))
                        GOTO(ERR1, ret);

                gc_ctx->gc_count++;
                DINFO_NOP("gc_count %llu\n", (LLU )gc_ctx->gc_count);
                if (gc_ctx->gc_count % LSV_GC_FULLGC_INTERVAL == 0) {
                        // lsv_log_fullgc_offer(lsv_info);
                }
        }

        return 0;
ERR1:
        yfree((void **) &ctx);
ERR0:
        return ret;
}

static int __do_fullgc(void *arg) {
        int ret;
        lsv_volume_proto_t *lsv_info = arg;

        ret = lsv_gc_fullgc(lsv_info);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int lsv_gc_fullgc_offer(lsv_volume_proto_t *lsv_info) {
        int ret;
        lsv_log_info_t *log_info = lsv_info->log_info;
        lsv_gc_context_t *gc_ctx = log_info->gc_context;

        if (lsv_feature & LSV_FEATURE_GC) {
                ret = co_mq_offer(&gc_ctx->fullgc_queue, "log_fullgc", __do_fullgc, (void *) lsv_info, 0);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static void __do_gc_add_to_check_queue(void *arg) {
        int rc = 0;
        lsv_log_proto_t log_ptoto;
        lsv_gc_add2check_arg_t * add2check_arg = arg;
        lsv_u32_t *task_count;

        log_ptoto.log = NULL;
        log_ptoto.chunk_id = add2check_arg->chunk_id;
        task_count = add2check_arg->count;
        rc = lsv_gc_add_to_check_queue(add2check_arg->lsv_info, &log_ptoto, add2check_arg->add_type);
        if (rc < 0) {
                DERROR("fullgc add log err,errno:%d\n,chunk_id:%u\n", rc, log_ptoto.chunk_id);
        }

        DINFO("fullgc add log, chunk_id %u\n", log_ptoto.chunk_id);
        *task_count = *task_count - 1;

        if (*task_count == 0) {
                schedule_resume(add2check_arg->wait_task, 0, NULL);
        }

        if (add2check_arg->is_free) {
                free(add2check_arg);
        }
}

int lsv_gc_add_to_check_queue_batch(lsv_volume_proto_t *lsv_info, lsv_u32_t chunk_count, lsv_u32_t *chunk_id_array,
                lsv_u8_t add_type) {

        lsv_log_info_t *log_info = NULL;
        lsv_gc_context_t *gc_context = NULL;

        log_info = lsv_info->log_info;
        gc_context = log_info->gc_context;

        if (gc_context->bitmap_is_open) {
                lsv_gc_add2check_arg_t * add2check_arg;
                task_t wait_task;
                lsv_u32_t task_count;
                lsv_u32_t i;

                if (chunk_count == 0) {
                        return 0;
                }

                task_count = 0;
                wait_task = schedule_task_get();

                for (i = 0; i < chunk_count; i++) {
                        task_count++;

                        add2check_arg = (lsv_gc_add2check_arg_t *) malloc(sizeof(lsv_gc_add2check_arg_t));
                        add2check_arg->lsv_info = lsv_info;
                        add2check_arg->chunk_id = chunk_id_array[i];
                        add2check_arg->add_type = add_type;
                        add2check_arg->count = &task_count;
                        add2check_arg->wait_task = &wait_task;
                        add2check_arg->is_free = 1;

                        schedule_task_new("__do_gc_add_to_check_queue", __do_gc_add_to_check_queue, add2check_arg, -1);
                }

                DINFO("lsv_log_gc_add_to_check_queue_batch add task. chunk_count %u\n", chunk_count);

                if (task_count > 0) {
                        schedule_yield("lsv_log_gc_add_to_check_queue_batch", NULL, NULL);
                }

                DINFO("lsv_log_gc_add_to_check_queue_batch add log. chunk_count %u\n", chunk_count);

                return 0;
        } else {
                int ret = 0;
                lsv_log_proto_t log_ptoto;

                for (uint32_t i = 0; i < chunk_count; i++) {
                        log_ptoto.log = NULL;
                        log_ptoto.chunk_id = chunk_id_array[i];
                        ret = lsv_gc_add_to_check_queue(lsv_info, &log_ptoto, add_type);
                        if (ret < 0) {
                                DERROR("fullgc add log err, ret %d\n chunk_id %u\n", ret, log_ptoto.chunk_id);
                        }
                }
                return 0;
        }
}
